package com.zhike.app.dim;

import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.zhike.bean.TableProcess;
import com.zhike.function.PhoenixSink;
import com.zhike.function.TableProcessFunction;
import com.zhike.utils.KafkaUtils;
import com.zhike.utils.MysqlUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Author: zhike
 * @CreateTime: 2024/1/17
 * @Description: Dim层开发
 *
 */
public class DimApp {
    public static void main(String[] args) {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度，一般与kafka的分区一致
        env.setParallelism(1);

        //2.读取kafka中car_db主题的数据创建主流
        KafkaSource<String> kafkaSource = KafkaUtils.getKafkaSource("car_db","DimApp");
        //创建主流
        DataStreamSource<String> stream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");

        //3.过滤非json数据并将数据转换为json格式
        SingleOutputStreamOperator<JSONObject> filterJsonDS = stream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    //如果不是json数据就会抛异常
                    JSONObject jsonObject = JSONObject.parseObject(value);
                        out.collect(jsonObject);
                } catch (Exception e) {
                    System.out.println("发现脏数据:" + value);
                }
            }
        });

        //4.使用flink cdc读取配置表创建配置流
        MySqlSource<String> mysqlSource = MysqlUtils.getMysqlSource("car_data.table_process");
        DataStreamSource<String> cdcSourceDS = env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "cdc_source");

        //5.将配置流处理为广播流
        MapStateDescriptor<String, TableProcess> mapStateDescriptor = new MapStateDescriptor<>("map_state", String.class, TableProcess.class);
        BroadcastStream<String> broadcast = cdcSourceDS.broadcast(mapStateDescriptor);

        //5.连接主流与广播流
        BroadcastConnectedStream<JSONObject, String> connectDS = filterJsonDS.connect(broadcast);

        //6.处理连接流
        SingleOutputStreamOperator<JSONObject> processDS = connectDS.process(new TableProcessFunction(mapStateDescriptor));

//        processDS.print();
        //7.将数据写入Phoenix表
        processDS.addSink(new PhoenixSink());

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

}
